package com.web.webtitle.utils;

import com.web.webtitle.entity.TitleHot;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;

@Component
public class HBaseUtils {
    //HBase连接对象
    private final Connection connection;
    //一旦创建对象 此方法就会加载
    public HBaseUtils() throws IOException {
        //设置本机客户端访问路径
        System.setProperty("hadoop.home.dir","E:\\hadoop");
        //设置访问身份
        System.setProperty("HADOOP_USER_NAME","root");
        //获取配置对象
        Configuration config = HBaseConfiguration.create();
        //hbase客户端配置zookeeper的集群地址
        config.set("hbase.zookeeper.quorum", "master,slave1,slave2");
        //设置端口号
        config.set("hbase.zookeeper.property.clientPort", "2181");
        //获取连接对象
        connection= ConnectionFactory.createConnection(config);
    }

    /**
     * 创建表
     * @param tableName 表名
     * @param family 列族
     * @throws IOException
     */
    public void createTable(String tableName,String...family) throws IOException {
        try {
            //获取操作对象
            Admin admin = connection.getAdmin();
            //创建表
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
            for (String s : family) {
                tableDescriptorBuilder.setColumnFamily( ColumnFamilyDescriptorBuilder.of(s));
            }
            //创建表
            admin.createTable(tableDescriptorBuilder.build());
            //关闭连接
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 新增数据
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifierName
     * @param qualifierValue
     * @throws IOException
     */
    public  void put(String tableName,String rowKey,String family,String qualifierName,String qualifierValue) throws IOException {
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf(tableName));
        //创建put的集合
        List<Put> putList = new ArrayList<>();
        putList.add( new Put(Bytes.toBytes(rowKey)).addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifierName), Bytes.toBytes(qualifierValue)));
        //提交数据
        table.put(putList);
        //关闭
        table.close();
        connection.close();
    }

    /**
     * 查询行数据
     * @param tableName
     * @param rowKey
     * @return
     * @throws IOException
     */
    public  Map<String,String>  get(String tableName,String rowKey) throws IOException{
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf(tableName));
        //结果集
        Result result = table.get(new Get(Bytes.toBytes(rowKey)));
        //map装载数据
        Map<String, String> rowInfo = new HashMap<>();
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            rowInfo.put("timestamp", Long.toString(cell.getTimestamp()));
            rowInfo.put("rowKey", Bytes.toString(CellUtil.cloneRow(cell)));
            rowInfo.put("family", Bytes.toString(CellUtil.cloneFamily(cell)));
            rowInfo.put("qualifier", Bytes.toString(CellUtil.cloneQualifier(cell)));
            rowInfo.put("qualifierValue", Bytes.toString(CellUtil.cloneValue(cell)));
        }
        //关闭
        table.close();
        connection.close();
    return rowInfo;
    }

    /**
     * 查询所有表
     * @param tableName
     * @param startTime
     * @param endTime
     * @return
     * @throws IOException
     */
    public  Map<String, Object>  scan(String tableName,Long startTime,Long endTime) throws IOException{
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf(tableName));
        //获取扫描对象
        Scan scan = new Scan();
        //得到某一个时间段的数据
        scan.setTimeRange(startTime,endTime);
        //获取迭代器
        Iterator<Result> iterator = table.getScanner(scan).iterator();
        //怎么装？ Map<String,List<Map<String,String>>>
        Map<String, Object> allData = new HashMap<>();
        //遍历
        while (iterator.hasNext()){
            Map<String, String> qualifierData = new HashMap<>();
            //获取下一行的数据
            Result result = iterator.next();//行数据
            //获取行键
            byte[] row = result.getRow();
            //单元格
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                qualifierData.put( Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
            }
            //装载行信息
            allData.put(new String(row),qualifierData);
        }
        table.close();
        connection.close();
        return allData;
    }

    /**
     * 使用对象存储热点话题
     * @param tableName
     * @param startTime
     * @param endTime
     * @return
     * @throws IOException
     */
    public  List<TitleHot> scanWithList(String tableName, Long startTime, Long endTime) throws IOException{
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf(tableName));
        //获取扫描对象
        Scan scan = new Scan();
        //得到某一个时间段的数据
        scan.setTimeRange(startTime,endTime);
        //获取迭代器
        Iterator<Result> iterator = table.getScanner(scan).iterator();
        List<TitleHot> titleHots = new ArrayList<>();
        //遍历
        while (iterator.hasNext()){
            //获取下一行的数据
            Result result = iterator.next();//行数据
            //单元格
            List<Cell> cells = result.listCells();
            TitleHot hot = new TitleHot();
            for (Cell cell : cells) {
                hot.setId(Bytes.toString(CellUtil.cloneRow(cell)));
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                if ("userName".equals(col)){
                    hot.setUserName(value);
                }else if ("agentType".equals(col)){
                    hot.setAgentType(value);
                }else if ("time".equals(col)){
                    hot.setTime(value);
                }else if ("title".equals(col)){
                    hot.setTitle(value);
                }

            }
            //装载行信息
            titleHots.add(hot);
        }
        table.close();
        connection.close();
        return titleHots;
    }

    /**
     * 删除列
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifierName
     * @throws IOException
     */
    public void delete(String tableName,String rowKey,String family,String qualifierName) throws IOException {
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf(tableName));
        //获取delete对象
        Delete delete = new Delete(rowKey.getBytes());
        //添加删除列
        delete.addColumn(family.getBytes(),qualifierName.getBytes());
        //删除
        table.delete(delete);
        table.close();
        connection.close();
    }

    /**
     * 删除表
     * @param tableName
     * @throws Exception
     */
    public void drop(String tableName) throws Exception {
        //获取操作对象
        Admin admin = connection.getAdmin();
        //删除表之前先停用表
        admin.disableTable(TableName.valueOf(tableName));
        admin.deleteTable(TableName.valueOf(tableName));
        admin.close();
        connection.close();
    }
}
